package com.hmy.mq.kafka.consumer;

import com.hmy.mq.kafka.constant.TopicNameConstant;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

@Component
public class KafkaConsumerListener {

//    @KafkaListener(topics = TopicNameConstant.TOPIC_ONE)
//    public void consumer1(String msg) {
//        System.out.printf("接收到来自topic:%s的消息，内容：[%s]%n", TopicNameConstant.TOPIC_ONE, msg);
//    }

    /**
     * 手动提交offset
     * @param record
     * @param ack
     */
    @KafkaListener(topics = {TopicNameConstant.TOPIC_ONE})
    public void initiativeCommitOffset1(ConsumerRecord<String, String> record, Acknowledgment ack) {
        try {
            System.out.printf("----- topic is %s, offset is %d,partition is %s, value is %s ----- \n", record.topic(), record.offset(),record.partition(), record.value());
            // 手动提交offset
            ack.acknowledge();
        }catch (Exception e){
            e.printStackTrace();
        }
    }

}
